package com.loong.spark.scala.streaming

import org.apache.log4j.{Level, Logger}
import org.apache.spark.SparkConf
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{Minutes, Seconds, StreamingContext}

/**
  * Created by 梁浩峰 on 2016/9/12 23:17.
  */
object KafkaWordCount {
  def main(args: Array[String]): Unit = {
    if(args.length<4){
      System.err.println("Usage: KafkaWordCount <zkQuorum> <group> <topics> <numThreads>")
      System.exit(1)
    }

    setStreamingLogLevels()

    val Array(zkQuorum,group,topics,numThreads)=args
    val sparkConf = new SparkConf().setAppName("KafkaWordCount").setMaster("local[2]")
    val ssc=new StreamingContext(sparkConf,Seconds(2))
    ssc.checkpoint("checkpoint")

    val topicMap = topics.split(",").map((_, numThreads.toInt)).toMap
    val lines = KafkaUtils.createStream(ssc,zkQuorum,group,topicMap).map(_._2)


    val words = lines.flatMap(_.split(" "))
    val wordCounts=words.map(x =>(x,1L)).reduceByKeyAndWindow(_+_,_-_,Minutes(10),Seconds(2),2)

    wordCounts.print()

    ssc.start()
    ssc.awaitTermination()
  }

  def setStreamingLogLevels(): Unit ={
    val log4jInitialized = Logger.getRootLogger.getAllAppenders.hasMoreElements
    if(!log4jInitialized){
      Logger.getRootLogger.setLevel(Level.WARN)
    }
  }

}
